package com.tid.hbasees.repo;

import com.google.common.base.Optional;
import com.google.common.collect.Lists;
import com.tid.hbasees.model.Node;
import com.tid.hbasees.utils.hbase.HBaseHelper;
import com.tid.hbasees.utils.JsonHelper;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.log4j.Logger;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.List;

@Component
public class NodeRepo {

    private static final Logger LOG = Logger.getLogger(NodeRepo.class);

    public static String TABLE_NAME = "node";
    public static String INDEX_TABLE_NAME = TABLE_NAME + "_index";

    @Resource(name = "hbaseHelper")
    private HBaseHelper hbaseHelper;

    @Resource
    private JsonHelper jsonHelper;

    @Resource
    private DeviceRepo deviceRepo;

    private HTableInterface getTable() throws IOException {
        return hbaseHelper.getTable(TABLE_NAME);
    }

    public List<Node> getNodesAndDevices(String nodeId, long startTime, long endTime) {
        return getNodes(nodeId, startTime, endTime, true);
    }

    public List<Node> getNodes(String nodeId, long startTime, long endTime) {
        return getNodes(nodeId, startTime, endTime, false);
    }

    public List<Node> getNodes(long startTime, long endTime) {
        List<Node> result = Lists.newLinkedList();

        Scan s = hbaseHelper.buildTypicalScan(startTime, endTime);

        try (HTableInterface table = getTable(); ResultScanner rs = table.getScanner(s)) {

            for (Result r : rs) {
                String json = hbaseHelper.getJsonContent(r);
                if (json != null) {
                    Optional<Node> nodeOptional = jsonHelper.readValue(json, Node.class);
                    if (nodeOptional.isPresent()) {
                        result.add(nodeOptional.get());
                    }
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
            LOG.error("failed to get nodes with condition: " + startTime + " " + endTime);
        }
        return result;
    }

    private List<Node> getNodes(String nodeId, long startTime, long endTime, boolean withDevices) {
        List<Node> result = Lists.newLinkedList();
        if (StringUtils.isEmpty(nodeId)) return result;

        Scan s = hbaseHelper.buildTypicalScan(nodeId, startTime, endTime);

        try (HTableInterface table = getTable(); ResultScanner rs = table.getScanner(s)) {

            for (Result r : rs) {
                String json = hbaseHelper.getJsonContent(r);
                if (json != null) {
                    Optional<Node> nodeOptional = jsonHelper.readValue(json, Node.class);
                    if (nodeOptional.isPresent()) {
                        Node n = nodeOptional.get();
                        if (withDevices) {
                            n.setDevices(deviceRepo.getByNodeId(nodeId, startTime, endTime));
                        }
                        result.add(nodeOptional.get());
                    }
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
            LOG.error("failed to get nodes with condition: " + startTime + " " + endTime + " " + nodeId);
        }
        return result;
    }

}
